package cn.xinfei.xdecision.data.core.frame.dispatch;

import cn.xinfei.xdecision.common.model.datax.consts.VariableConst;
import cn.xinfei.xdecision.common.model.datax.datainterface.v2.VarInterfaceInfo;
import cn.xinfei.xdecision.common.model.datax.datamanage.v2.Variable;
import cn.xinfei.xdecision.common.model.datax.datamanage.v2.VariableCallLog;
import cn.xinfei.xdecision.common.model.datax.enums.CollectStageEnum;
import cn.xinfei.xdecision.common.model.datax.enums.ExecutorEnum;
import cn.xinfei.xdecision.common.model.datax.enums.ExecutorSceneEnum;

import cn.xinfei.xdecision.common.model.datax.enums.VariableEnum;

import cn.xinfei.xdecision.common.model.datax.exception.XDecisionDataException;
import cn.xinfei.xdecision.common.model.datax.exception.XDecisionDataExceptionType;
import cn.xinfei.xdecision.common.service.datax.datainterface.v2.VarInterfaceInfoService;
import cn.xinfei.xdecision.common.service.datax.datainterface.v2.VariableInterfaceService;

import cn.xinfei.xdecision.common.service.datax.datamanage.v2.VariableService;
import cn.xinfei.xdecision.data.core.cache.VarInterfaceCache;
import cn.xinfei.xdecision.data.core.cache.VarMetaDataCache;
import cn.xinfei.xdecision.data.core.config.ThreadPoolConfig;
import cn.xinfei.xdecision.data.core.frame.executors.groovy.GroovyExecutorService;
import cn.xinfei.xdecision.data.core.frame.executors.http.HttpStandardExecutorService;
import cn.xinfei.xdecision.data.core.frame.executors.http.HttpVirtualExecutorService;
import cn.xinfei.xdecision.data.core.frame.executors.sql.DbSqlExecutorService;
import cn.xinfei.xdecision.data.core.frame.listener.BatchSaveVarLogEvent;
import cn.xinfei.xdecision.data.core.util.VariableFutureUtils;
import cn.xinfei.xdecision.data.core.vo.*;
import cn.xinfei.xdecision.data.service.VariableCallLogService;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.google.common.collect.Lists;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.stream.Collectors;

@Service
public class ExecutorDispatchService implements ExecutorDispatch {

    private static final Logger LOGGER = LoggerFactory.getLogger(ExecutorDispatchService.class);

    private static final String HTTP_CONFIG = "HTTP";
    private static final String GROOVY_PRE = "GROOVY_PRE";
    private static final String GROOVY_AFTER = "GROOVY_AFTER";
    private static final String SQL = "SQL";
    private static final String VIRTUAL = "VIRTUAL";
    private static final String STANDARD = "STANDARD";

    private static final int CORE_POOL_SIZE = 5;
    private static final int QUEUE_SIZE = CORE_POOL_SIZE * 1000;

    @Autowired
    ThreadPoolConfig threadPoolConfig;
    @Autowired
    HttpStandardExecutorService httpStandardExecutorService;
    @Autowired
    HttpVirtualExecutorService httpVirtualExecutorService;
    @Autowired
    GroovyExecutorService groovyExecutorService;
    @Autowired
    DbSqlExecutorService dbSqlExecutorService;
    @Autowired
    VariableService variableService;
    @Autowired
    VariableInterfaceService variableInterfaceService;
    @Autowired
    VarInterfaceInfoService varInterfaceInfoService;
    @Autowired
    VariableCallLogService variableCallLogService;

    @Autowired
    ApplicationContext applicationContext;

    @Resource(name = "standardThreadPoolExecutor")
    ExecutorService httpStandardExecutor;

    @Resource(name = "virtualThreadPoolExecutor")
    ExecutorService httpVirtualExecutor;

    @Resource(name = "groovyThreadPoolExecutor")
    ExecutorService groovyExecutor;

    @Autowired
    VarInterfaceCache varInterfaceCache;

    @Autowired
    VarMetaDataCache varMetaDataCache;


    @Value("${xdecision.data.executor.timeout.http_standard}")
    private int httpStandardExecutorTimeout;

    @Value("${xdecision.data.executor.timeout.http_virtual}")
    private int httpVirtualExecutorTimeout;

    @Value("${xdecision.data.executor.timeout.groovy}")
    private int groovyExecutorTimeout;

    @Override
    public void dispatch(String requestId,
                         Set<String> varCodes,
                         JSONObject inputPrarams,
                         Map<String, Set<String>> executorAndVarCodesMap,
                         Map<String, Set<String>> varDependParamMap,
                         Map<String, Object> dispatchResult,
                         CollectStageEnum stage) {
        Set<String> httpVars = executorAndVarCodesMap.get(ExecutorEnum.HTTP.name());
        Set<String> virtualHttpVars = Collections.emptySet();
        Set<String> standardHttpVars = Collections.emptySet();

        Optional<Map<Integer, VarInterfaceInfo>> varInterfaceDataCacheLoader = Optional.ofNullable(varInterfaceCache.loadVarInterfaceMetaData());
        Optional<Map<String, Variable>> varMetaDataCache = Optional.ofNullable(this.varMetaDataCache.loadVarMetaData());
        Map<String, Variable> virtualVarCache = new HashMap<>();
        Map<String, Variable> standardVarCache = new HashMap<>();


        if (CollectionUtils.isNotEmpty(httpVars)) {
            if (varMetaDataCache.isPresent() && varInterfaceDataCacheLoader.isPresent()) {
                LOGGER.info("requestId = {} ExecutorDispatch 查询缓存 VarMetaDataCache & VarInterfaceDataCache", requestId);
                virtualVarCache = buildInterfaceVarCache(varInterfaceDataCacheLoader, varMetaDataCache, VariableConst.VIRTUAL_DATASOURCE);
                standardVarCache = buildInterfaceVarCache(varInterfaceDataCacheLoader, varMetaDataCache, VariableConst.STANDARD_DATASOURCE);

                Map<String, Variable> finalVirtualVarCache = virtualVarCache;
                virtualHttpVars = httpVars.stream().filter(finalVirtualVarCache::containsKey)
                        .collect(Collectors.toSet());
                standardHttpVars = new HashSet<>(CollectionUtils.subtract(httpVars, virtualHttpVars));

            } else {
                virtualHttpVars = httpVars.stream().filter(item -> variableService.checkVirtualDataSource(item))
                        .collect(Collectors.toSet());
                standardHttpVars = new HashSet<>(CollectionUtils.subtract(httpVars, virtualHttpVars));
            }
        }


        Set<String> preGroovyVars = executorAndVarCodesMap.get(ExecutorEnum.GROOVY_PRE.name());
        Set<String> afterGroovyVars = executorAndVarCodesMap.get(ExecutorEnum.GROOVY_AFTER.name());

        Set<String> finalVirtualVars = virtualHttpVars;
        Set<String> finalStandardVars = standardHttpVars;

        Map<Long, Set<String>> virtualDataSourceIdAndVarCodesMap = buildDataSourceAndCodesMap(virtualVarCache, finalVirtualVars);
        Map<Long, Set<String>> standardDataSourceIdAndVarCodesMap = buildDataSourceAndCodesMap(standardVarCache, finalStandardVars);
        List<Long> virtualHttpDataSourceIdList = new ArrayList<>(Optional.ofNullable(virtualDataSourceIdAndVarCodesMap)
                .orElse(Collections.emptyMap())
                .keySet());
        List<Long> standardDataSourceIdList = new ArrayList<>(Optional.ofNullable(standardDataSourceIdAndVarCodesMap)
                .orElse(Collections.emptyMap())
                .keySet());

        LOGGER.info("requestId = {} httpVars: {},standardHttpVars:{}, virtualHttpVars: {}, preGroovyVars: {}",
                requestId,
                JSONObject.toJSONString(httpVars),
                JSONObject.toJSONString(standardHttpVars),
                JSONObject.toJSONString(virtualHttpVars),
                JSONObject.toJSONString(preGroovyVars));
        List<ExecutorEnum> executorEnumList = executorAndVarCodesMap.keySet().stream().map(ExecutorEnum::valueOf).collect(Collectors.toList());

        List<Future<VirtualHttpExecutorResultEntity>> virtualHttpFutureList = new ArrayList<>();
        List<Future<StandardHttpExecutorResultEntity>> standardHttpFutureList = new ArrayList<>();
        List<Future<GroovyExecutorResultEntity>> groovyPreFutureList = new ArrayList<>();
        List<Future<GroovyExecutorResultEntity>> groovyAfterFutureList = new ArrayList<>();

        Set<String> finalVirtualHttpVars = virtualHttpVars;
        Set<String> finalStandardHttpVars = standardHttpVars;
        executorEnumList.forEach(executorType -> {
            switch (executorType) {
                case HTTP:
                    if (CollectionUtils.isNotEmpty(finalVirtualHttpVars)) {
                        LOGGER.info("requestId = {} [VirtualHttpExecutor] start execute!", requestId);
                        virtualHttpDataSourceIdList.forEach(datasourceId -> {
                            Set<String> virtualVarCodes = virtualDataSourceIdAndVarCodesMap.get(datasourceId);
                            virtualHttpFutureList.add(httpVirtualExecutor.submit(() -> {
                                ConcurrentHashMap<String, Object> inputParamMap = new ConcurrentHashMap<>(inputPrarams);
                                HttpVirtualExecutorTask httpVirtualExecutorTask = new HttpVirtualExecutorTask(
                                        virtualVarCodes,
                                        datasourceId,
                                        inputParamMap,
                                        httpVirtualExecutorService,
                                        stage,
                                        requestId);
                                return httpVirtualExecutorTask.call();
                            }));
                        });
                        LOGGER.info("requestId = {} [VirtualHttpExecutor] wait...", requestId);
                    }

                    if (CollectionUtils.isNotEmpty(finalStandardHttpVars)) {
                        LOGGER.info("requestId = {} [StandardHttpExecutor] start execute!", requestId);
                        standardDataSourceIdList.forEach(datasourceId -> {
                            Set<String> standardVarCodes = standardDataSourceIdAndVarCodesMap.get(datasourceId);
                            standardHttpFutureList.add(httpStandardExecutor.submit(() -> {
                                ConcurrentHashMap<String, Object> inputParamMap = new ConcurrentHashMap<>(inputPrarams);
                                HttpStandardExecutorTask standardExecutorTask = new HttpStandardExecutorTask(
                                        standardVarCodes,
                                        datasourceId,
                                        inputParamMap,
                                        httpStandardExecutorService,
                                        stage,
                                        requestId);
                                return standardExecutorTask.call();
                            }));
                        });
                        LOGGER.info("requestId = {} [StandardHttpExecutor] wait...", requestId);
                    }
                    break;

                case GROOVY_PRE:
                    LOGGER.info("requestId = {} [GroovyPreExecutor] start execute!", requestId);
                    preGroovyVars.forEach(groovyVar -> {
                        groovyPreFutureList.add(groovyExecutor.submit(() -> {
                            ConcurrentHashMap<String, Object> inputParamMap = new ConcurrentHashMap<>(inputPrarams);
                            GroovyExecutorTask groovyExecutorTask = new GroovyExecutorTask(
                                    groovyVar,
                                    ExecutorEnum.GROOVY_PRE,
                                    inputParamMap,
                                    groovyExecutorService,
                                    null,
                                    varDependParamMap,
                                    stage,
                                    requestId);
                            return groovyExecutorTask.call();
                        }));
                    });
                    LOGGER.info("requestId = {} [GroovyPreExecutor] start wait...", requestId);
                    break;

                case GROOVY_AFTER:
                case SQL:
                    break;
                default:
                    throw new XDecisionDataException(XDecisionDataExceptionType.DATA_NO_SUCH_EXECUTOR_ERROR,
                            "no such executor type: " + executorType.name());
            }
        });
//        List<VariableCallLog> callLogs = Lists.newArrayList();
        try {
            List<VirtualHttpExecutorResultEntity> virtualHttpResultList = VariableFutureUtils.getExecutorFutures(VIRTUAL, virtualHttpFutureList, httpVirtualExecutorTimeout);
            List<StandardHttpExecutorResultEntity> standardHttpResultList = VariableFutureUtils.getExecutorFutures(STANDARD, standardHttpFutureList, httpStandardExecutorTimeout);
            List<GroovyExecutorResultEntity> groovyPreResultList = VariableFutureUtils.getExecutorFutures(GROOVY_PRE, groovyPreFutureList, groovyExecutorTimeout);
            List<GroovyExecutorResultEntity> groovyAfterResultList = new ArrayList<>();
            LOGGER.info("requestId = {}, virtualHttpResultList: {}, standardHttpResultList:{}, groovyPreResultList:{}",
                    requestId,
                    JSONObject.toJSONString(virtualHttpResultList),
                    JSONObject.toJSONString(standardHttpResultList),
                    JSONObject.toJSONString(groovyPreResultList)
            );

            Map<String, Object> virtualResult = VariableFutureUtils.flatFutureMap(virtualHttpResultList);
            Map<String, Object> standardResult = VariableFutureUtils.flatFutureMap(standardHttpResultList);
            Map<String, Object> groovyPreResult = VariableFutureUtils.flatFutureMap(groovyPreResultList);
            LOGGER.info("requestId = {}, flatVirtualResult: {}, flatStandardResult:{}, flatGroovyPreResult:{}",
                    requestId,
                    JSONObject.toJSONString(virtualResult),
                    JSONObject.toJSONString(standardResult),
                    JSONObject.toJSONString(groovyPreResult)
            );


            dispatchResult.putAll(virtualResult);
            dispatchResult.putAll(standardResult);
            dispatchResult.putAll(groovyPreResult);

            if (CollectionUtils.isNotEmpty(afterGroovyVars)) {
                LOGGER.info("requestId = {} [GroovyAfterExecutor] start execute! afterGroovyVars: {}", requestId, JSONObject.toJSONString(afterGroovyVars));
                afterGroovyVars.forEach(groovyVar -> {
                    groovyAfterFutureList.add(groovyExecutor.submit(() -> {
                        ConcurrentHashMap<String, Object> inputParamMap = new ConcurrentHashMap<>(inputPrarams);
                        GroovyExecutorTask groovyExecutorTask = new GroovyExecutorTask(groovyVar,
                                ExecutorEnum.GROOVY_AFTER,
                                inputParamMap,
                                groovyExecutorService,
                                dispatchResult,
                                varDependParamMap,
                                stage,
                                requestId);
                        return groovyExecutorTask.call();
                    }));
                });
                LOGGER.info("requestId = {} [GroovyAfterExecutor] start wait...", requestId);

                try {
                    groovyAfterResultList = VariableFutureUtils.getExecutorFutures(GROOVY_AFTER, groovyAfterFutureList, groovyExecutorTimeout);
                    Map<String, Object> groovyAfterFlatResult = VariableFutureUtils.flatFutureMap(groovyAfterResultList);
                    LOGGER.info("requestId = {} groovyAfterResultList: {} flatGroovyAfterResult: {}",
                            requestId,
                            JSONObject.toJSONString(groovyAfterResultList),
                            JSONObject.toJSONString(groovyAfterFlatResult));

                    dispatchResult.putAll(groovyAfterFlatResult);
                } catch (XDecisionDataException e) {
                    LOGGER.error("requestId = {} error in groovyAfter Executors, executorAndVarCodesMap:{}", requestId, JSONObject.toJSONString(executorAndVarCodesMap), e);
                    throw new XDecisionDataException(XDecisionDataExceptionType.DATA_EXECUTOR_ERROR, "执行器执行异常!", e);
                }

            } else {
                LOGGER.info("requestId = {} afterGroovyVars is empty!", requestId);
            }
            LOGGER.info("requestId = {} dispatch result = {}", requestId, JSONObject.toJSONString(dispatchResult, SerializerFeature.WRITE_MAP_NULL_FEATURES, SerializerFeature.QuoteFieldNames));

//            List<VariableCallLog> virtualHttpCallLogs = VariableFutureUtils.flatCallLogs(virtualHttpResultList);
//            List<VariableCallLog> standardHttpCallLogs = VariableFutureUtils.flatCallLogs(standardHttpResultList);
//            List<VariableCallLog> groovyPreCallLogs = VariableFutureUtils.flatCallLogs(groovyPreResultList);
//            List<VariableCallLog> groovyAfterCallLogs = VariableFutureUtils.flatCallLogs(groovyAfterResultList);

//            callLogs.addAll(virtualHttpCallLogs);
//            callLogs.addAll(standardHttpCallLogs);
//            callLogs.addAll(groovyPreCallLogs);
//            callLogs.addAll(groovyAfterCallLogs);
//
//            BatchSaveVarDTO batchSaveVarDTO = new BatchSaveVarDTO();
//            batchSaveVarDTO.setRequestId(requestId);
//            batchSaveVarDTO.setVarCallLogList(callLogs);
//            batchSaveVarDTO.setCollectStage(stage.name());

            //异步记录变量日志
//            applicationContext.publishEvent(new BatchSaveVarLogEvent(this, batchSaveVarDTO));

        } catch (RejectedExecutionException ex) {
            LOGGER.error("requestId = {}task executing was rejected, executorAndVarCodesMap = {}", requestId, JSONObject.toJSONString(executorAndVarCodesMap), ex);
        } catch (Exception e) {
            LOGGER.error("requestId = {} error in dispatch Executors,executorAndVarCodesMap = {}", requestId, JSONObject.toJSONString(executorAndVarCodesMap), e);
        }

    }

    private Map<String, Variable> buildInterfaceVarCache(Optional<Map<Integer, VarInterfaceInfo>> varInterfaceDataCacheLoader,
                                                         Optional<Map<String, Variable>> varMetaDataCache,
                                                         final Map<String, String> dataSourceTypeMap) {
        Map<String, Variable> virtualVarCache;
        virtualVarCache = varMetaDataCache.get()
                .values()
                .stream()
                .filter(var -> var.getVarType().equals(Integer.parseInt(VariableEnum.INTERFACE_VAR.getCode())))
                .filter(var -> {
                    Optional<VarInterfaceInfo> interfaceInfo = varInterfaceDataCacheLoader.get().values().stream()
                            .filter(inter -> var.getInterfaceId().equals(Long.valueOf(inter.getId())))
                            .filter(inter -> dataSourceTypeMap.containsKey(inter.getType()))
                            .findFirst();
                    return interfaceInfo.isPresent();
                })
                .collect(Collectors.toMap(Variable::getVarCode, Function.identity(), (e1, e2) -> e1));
        return virtualVarCache;
    }

    private Map<Long, Set<String>> buildDataSourceAndCodesMap(Map<String, Variable> varCache, Set<String> finalVars) {
        return Optional.of(varCache).map(cache ->
                cache.values().stream()
                        .filter(var -> finalVars.contains(var.getVarCode()))
                        .collect(Collectors.groupingBy(Variable::getInterfaceId,
                                Collectors.mapping(Variable::getVarCode, Collectors.toSet())))
        ).orElseGet(Collections::emptyMap);
    }

    @Data
    @RequiredArgsConstructor
    public static class HttpStandardExecutorTask implements Callable<StandardHttpExecutorResultEntity> {

        private final Set<String> varCodes;
        private final Long datasourceId;
        private final Map<String, Object> inputParam;
        private final HttpStandardExecutorService httpStandardExecutorService;
        private final CollectStageEnum stage;
        private final String requestId;

        @Override
        public StandardHttpExecutorResultEntity call() {
            MDC.put(VariableConst.HANDLE_NAME, ExecutorSceneEnum.HTTPSTAND.name());
            MDC.put(VariableConst.COLLECT_STAGE, stage.name());
            MDC.put(VariableConst.REQUEST_ID, requestId);

            LOGGER.info("requestId = {} dataSourceId = {} HttpStandardExecutorTask start run", requestId, datasourceId.toString());
            return httpStandardExecutorService.run(varCodes, inputParam, datasourceId, stage, requestId);
        }
    }

    @Data
    @RequiredArgsConstructor
    public static class HttpVirtualExecutorTask implements Callable<VirtualHttpExecutorResultEntity> {
        private final Set<String> varCodes;
        private final Long datasourceId;
        private final Map<String, Object> inputParam;
        private final HttpVirtualExecutorService httpVirtualExecutorService;
        private final CollectStageEnum stage;
        private final String requestId;

        @Override
        public VirtualHttpExecutorResultEntity call() {
            MDC.put(VariableConst.HANDLE_NAME, ExecutorSceneEnum.HTTPVIRTUAL.name());
            MDC.put(VariableConst.COLLECT_STAGE, stage.name());
            MDC.put(VariableConst.REQUEST_ID, requestId);
            LOGGER.info("requestId = {} dataSourceId = {} HttpVirtualExecutorTask start run", requestId, datasourceId.toString());
            return httpVirtualExecutorService.run(varCodes, inputParam, datasourceId, stage, requestId);
        }
    }

    @Data
    @RequiredArgsConstructor
    public static class GroovyExecutorTask implements Callable<GroovyExecutorResultEntity> {

        private final String varCode;
        private final ExecutorEnum executorEnum;
        private final Map<String, Object> inputParam;
        private final GroovyExecutorService groovyExecutorService;
        private final Map<String, Object> executorResult;
        private final Map<String, Set<String>> deriveDependParamMap;
        private final CollectStageEnum stage;
        private final String requestId;

        @Override
        public GroovyExecutorResultEntity call() {
            MDC.put(VariableConst.HANDLE_NAME, ExecutorSceneEnum.GROOVY.name());
            MDC.put(VariableConst.COLLECT_STAGE, stage.name());
            MDC.put(VariableConst.REQUEST_ID, requestId);
            LOGGER.info("requestId = {} GroovyExecutorTask start run", requestId);
            return groovyExecutorService.run(varCode, inputParam, deriveDependParamMap, executorEnum, executorResult, stage, requestId);
        }
    }

    @Data
    @RequiredArgsConstructor
    public static class SqlExecutorTask implements Callable<SqlExecutorResultEntity> {

        private final String varCode;
        private final Map<String, Object> inputParam;
        private final Map<String, Object> executorResult;
        private final DbSqlExecutorService sqlExecutorService;
        private final CollectStageEnum stage;
        private final String requestId;

        @Override
        public SqlExecutorResultEntity call() {
            LOGGER.info("requestId = {} SqlExecutorTask start run", requestId);
            return sqlExecutorService.run(varCode, inputParam, null, null, executorResult, stage, requestId);
        }
    }

}
